/*
 *
 *  * Copyright (c) iwindplus Technologies Co., Ltd.2024-2030, All rights reserved.
 *
 *
 */

package com.iwindplus.base.kafka.service;

import com.iwindplus.base.kafka.domain.property.KafkaProperty;
import com.iwindplus.base.kafka.support.KafkaTopicAutoCreator;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.springframework.beans.factory.SmartInitializingSingleton;

/**
 * kafka动态注册.
 *
 * @author zengdegui
 * @since 2020/4/24
 */
@Slf4j
public class KafkaDynamicRegistry implements SmartInitializingSingleton {

    private AdminClient adminClient;

    private KafkaProperty kafkaProperty;

    public KafkaDynamicRegistry(AdminClient adminClient, KafkaProperty kafkaProperty) {
        this.adminClient = adminClient;
        this.kafkaProperty = kafkaProperty;
    }

    @Override
    public void afterSingletonsInstantiated() {
        if (Boolean.FALSE.equals(kafkaProperty.getEnabled())
            || Boolean.FALSE.equals(kafkaProperty.getEnabledDynamicRegister())) {
            return;
        }

        KafkaTopicAutoCreator.createTopicsIfAbsent(adminClient, kafkaProperty.getConfigs(), 10L);
    }
}
